package com.diy.sigmund.diyorder.common.middleware.mq.listener;

import java.util.Arrays;
import java.util.List;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * @author snowalker
 * @version 1.0
 * @date 2019/6/15 11:25
 * @className SecKillChargeOrderListenerImpl
 * @desc 秒杀订单消费监听回调
 */
@Component
public class SecKillChargeOrderListenerImpl implements MessageListenerConcurrently {

    private static final Logger LOG = LoggerFactory.getLogger(SecKillChargeOrderListenerImpl.class);

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        try {
            for (MessageExt messageExt : msgs) {
                if (messageExt.getReconsumeTimes() == 3) {
                    // 重试消息的处理机制
                    // 一般情况下我们在实际生产中是不需要重试16次，这样既浪费时间又浪费性能，理论上当尝试重复次数
                    // 达到我们想要的结果时如果还是消费失败，那么我们需要将对应的消息进行记录，并且结束重复尝试
                    // 可以将对应的数据保存到数据库，以便人工干预
                    System.out.println(messageExt.getMsgId() + "," + Arrays.toString(messageExt.getBody()));
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }

                // 消息解码
                String message = new String(messageExt.getBody());
                int reconsumeTimes = messageExt.getReconsumeTimes();
                String msgId = messageExt.getMsgId();
                String logSuffix = ",msgId=" + msgId + ",reconsumeTimes=" + reconsumeTimes;
                LOG.info("[秒杀订单消费者]-SecKillChargeOrderConsumer-接收到消息,message={},{}", message, logSuffix);
                // RocketMQ提供了ack机制，以保证消息能够被正常消费。
                // 所有消费者在设置监听的时候会提供一个回调，业务实现消费回调的时候，当回调方法中返回
                // ConsumeConcurrentlyStatus.CONSUME_SUCCESS，RocketMQ才会认为这批消息（默认是1条）是
                // 消费完成的。如果这时候消息消费失败，例如数据库异常，余额不足扣款失败等一切业务认为消息需要
                // 重试的场景，只要返回ConsumeConcurrentlyStatus.RECONSUME_LATER，RocketMQ就会认为这批
                // 消息消费失败了
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        } catch (Exception e) {
            LOG.info("[秒杀订单消费者]消费异常", e);
        }
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
}
